# Databricks notebook source
# MAGIC %md
# MAGIC ## TMR Populate Task Dimensions
# MAGIC <p><b>Description: </b>This notebook aims to populate all task dimensions needed for the TMR reports A.K.A Developer Services.</p>
# MAGIC <b>Parent Process: </b>PL_FactPrism
# MAGIC <table class="table table-striped table-hover">
# MAGIC  <thead>
# MAGIC   <tr>
# MAGIC    <th>Contributor</th>
# MAGIC    <th>Date</th>
# MAGIC    <th>Version</th>
# MAGIC    <th>Comment</th>
# MAGIC    <th>WorkItem No</th>
# MAGIC   </tr>
# MAGIC  </thead>
# MAGIC  <tbody>
# MAGIC   <tr>
# MAGIC    <td>Iwan Dyke</td>
# MAGIC    <td>18/04/2024</td>
# MAGIC    <td>1.0</td>
# MAGIC    <td>Create initial release</td>
# MAGIC    <td></td>
# MAGIC   </tr>
# MAGIC  </tbody>
# MAGIC </table>

# COMMAND ----------

# MAGIC %md
# MAGIC ## Import Python Libraries

# COMMAND ----------

from pyspark.sql.types import *
from pyspark.sql.functions import *
from pyspark.sql.window import Window
import pandas as pd  
import numpy as np 
import os  # Joining filenames.
import datetime as dt  # Adding today's date to exports.
from datetime import datetime
import glob

# COMMAND ----------

# MAGIC %md
# MAGIC ## Populating Type 0 Task Dimensions

# COMMAND ----------

# Checking type 0 dimension dim_task_status

dim_task_status_sql_check = """
WITH CTE_dim_task_status_values AS (
SELECT DISTINCT 
  s.`Task Status`
FROM 
	databricks_poc.staging.tmr_raw s
	LEFT OUTER JOIN databricks_poc.tmr.dim_task_status d
	ON s.`Task Status` = d.SourceValue  
WHERE d.SourceValue IS NULL
)

SELECT ALL * FROM CTE_dim_task_status_values
"""

dim_task_status_data_check = spark.sql(dim_task_status_sql_check)

if dim_task_status_data_check.count() > 0:
  raise ValueError("New data found in dim_task_status")

# COMMAND ----------

# Checking type 0 dimension dim_lpa

dim_lpa_sql_check = """
WITH CTE_dim_lpa_values AS (
SELECT DISTINCT 
  s.`lpa`
FROM 
	databricks_poc.staging.tmr_raw s
	LEFT OUTER JOIN databricks_poc.tmr.dim_lpa d
	ON s.`lpa` = d.`Description`
WHERE d.Description IS NULL
)

SELECT ALL * FROM CTE_dim_lpa_values
WHERE lpa IS NOT NULL
"""

dim_lpa_data_check = spark.sql(dim_lpa_sql_check)

if dim_lpa_data_check.count() > 0:
  raise ValueError("New data found in dim_lpa")

# COMMAND ----------

# Checking type 0 dimension dim_task_status dim_bpm_prefix

dim_bpm_prefix_sql_check = """
WITH CTE_dim_bpm_prefix_values AS (
    SELECT DISTINCT 
        LEFT(s.`BPM Reference Number`, 3) AS Prefix
    FROM 
        databricks_poc.staging.tmr_raw s
        LEFT OUTER JOIN databricks_poc.tmr.dim_bpm_prefix d
        ON LEFT(s.`BPM Reference Number`, 3) = d.`Description`
    WHERE d.Description IS NULL
)
SELECT * FROM CTE_dim_bpm_prefix_values
"""

dim_bpm_prefix_data_check = spark.sql(dim_bpm_prefix_sql_check)

if dim_bpm_prefix_data_check.count() > 0:
    raise ValueError("New data found in dim_bpm_prefix")

# COMMAND ----------

# Checking type 0 dimension dim_process_status

dim_process_status_sql_check = """
WITH CTE_CleansedDimensionData AS (
SELECT
	CASE
		WHEN d.Description = 'Cancelled' THEN 'CANCELED'
		WHEN d.Description = 'In Progress' THEN 'IN_PROGRESS'
		WHEN d.Description = 'Completed' THEN 'COMPLETED'
		WHEN d.Description = 'Suspended' THEN 'SUSPENDED'
		ELSE d.Description 
	END AS Description
FROM 
	databricks_poc.tmr.dim_process_status d
)
,CTE_dim_process_status_values AS (
SELECT DISTINCT 
	s.`Root Process Status`
FROM 
	databricks_poc.staging.tmr_raw s
	LEFT OUTER JOIN CTE_CleansedDimensionData d
	ON s.`Root Process Status` = d.Description
WHERE 
	d.Description IS NULL
	AND s.`Root Process Status` <> ''
 )

SELECT ALL * FROM CTE_dim_process_status_values
"""

dim_process_status_data_check = spark.sql(dim_process_status_sql_check)

if dim_process_status_data_check.count() > 0:
  raise ValueError("New data found in dim_process_status")

# COMMAND ----------

# Checking type 0 dimension dim_process

dim_process_sql_check = """
WITH CTE_CleansedDimensionData
(
SELECT DISTINCT   
 SPLIT(`Current Sub Process Subject(0-60)`, ' : ')[1] AS Process
FROM   
  databricks_poc.staging.tmr_raw 
WHERE   
  LENGTH(`Current Sub Process Subject(0-60)`) > 0  
)
SELECT DISTINCT   
  Process 
FROM   
  CTE_CleansedDimensionData s 
LEFT OUTER JOIN 
  databricks_poc.tmr.dim_process d 
    ON s.Process = d.Description
WHERE   
  d.Description IS NULL;
"""

dim_process_data_check = spark.sql(dim_process_sql_check)

if dim_process_data_check.count() > 0:
  raise ValueError("New data found in dim_process")

# COMMAND ----------

# MAGIC %md
# MAGIC ## Populating Type 1 Task Dimensions

# COMMAND ----------

# MAGIC %sql
# MAGIC -- Check for new values in type 1 dimension dim_task_owner
# MAGIC
# MAGIC WITH CTE_dim_task_owner_cleansed AS 
# MAGIC (
# MAGIC SELECT DISTINCT 
# MAGIC   `TaskOwner`
# MAGIC FROM 
# MAGIC   databricks_poc.staging.tmr_raw s 
# MAGIC LATERAL VIEW EXPLODE(SPLIT(`Task Owner`, ',')) AS TaskOwner
# MAGIC )
# MAGIC ,CTE_dim_task_owner AS
# MAGIC (
# MAGIC   SELECT   
# MAGIC     LTRIM(`TaskOwner`) AS Description 
# MAGIC   FROM   
# MAGIC     CTE_dim_task_owner_cleansed s
# MAGIC   WHERE   
# MAGIC     LTRIM(`TaskOwner`) NOT LIKE 'test%'
# MAGIC     AND LTRIM(`TaskOwner`) NOT LIKE 'Test%'   
# MAGIC     AND LTRIM(`TaskOwner`) NOT LIKE '%...%' 
# MAGIC )
# MAGIC INSERT INTO databricks_poc.tmr.dim_task_owner
# MAGIC (
# MAGIC   `Description`
# MAGIC )
# MAGIC SELECT
# MAGIC   s.`Description`
# MAGIC FROM
# MAGIC   CTE_dim_task_owner s
# MAGIC LEFT OUTER JOIN databricks_poc.tmr.dim_task_owner d
# MAGIC   ON s.`Description` = d.`Description`
# MAGIC WHERE 
# MAGIC   d.`Description` IS NULL

# COMMAND ----------

# MAGIC %sql
# MAGIC -- Check for new values in type 1 dimension dim_designer
# MAGIC
# MAGIC WITH CTE_dim_designer_cleansed AS (
# MAGIC SELECT DISTINCT 
# MAGIC   s.`Designer First Name`
# MAGIC   ,s.`Designer Last Name`
# MAGIC   ,s.`Designer First Name` || ' ' || s.`Designer Last Name` AS FullName
# MAGIC FROM 
# MAGIC 	databricks_poc.staging.tmr_raw s
# MAGIC 	LEFT OUTER JOIN databricks_poc.tmr.dim_designer d
# MAGIC 	ON s.`Designer First Name` = d.`FirstName`
# MAGIC WHERE 
# MAGIC d.`FirstName` IS NULL
# MAGIC   AND s.`Designer First Name` <> ''
# MAGIC )
# MAGIC INSERT INTO databricks_poc.tmr.dim_designer
# MAGIC (
# MAGIC   `FirstName`
# MAGIC   ,`Lastname`
# MAGIC   ,`FullName`
# MAGIC )
# MAGIC SELECT
# MAGIC   `Designer First Name`
# MAGIC   ,`Designer Last Name`
# MAGIC   ,`FullName`
# MAGIC FROM
# MAGIC   CTE_dim_designer_cleansed

# COMMAND ----------

# MAGIC %md
# MAGIC ## Populating Type 2 Task Dimensions

# COMMAND ----------

# MAGIC %sql
# MAGIC --Dim_TaskSubject
# MAGIC -- Split up to first colon
# MAGIC CREATE OR REPLACE TEMPORARY VIEW StringSplit_Subject AS 
# MAGIC (
# MAGIC   SELECT 
# MAGIC     f.`Task Instance ID`,
# MAGIC     TRIM(x.`value`) AS value,
# MAGIC     row_number() OVER (PARTITION BY f.`Task Instance ID` ORDER BY f.`Task Instance ID`) AS RN
# MAGIC   FROM 
# MAGIC     databricks_poc.staging.tmr_raw f
# MAGIC   LATERAL VIEW explode(split(concat(f.`Task Subject(0-60)`, f.`Task Subject(61-120)`), ':')) x AS value
# MAGIC );
# MAGIC
# MAGIC -- Remove unwanted rows (excess text) following colon
# MAGIC CREATE OR REPLACE TEMPORARY VIEW StringSplit_Subject_II AS 
# MAGIC (
# MAGIC   SELECT 
# MAGIC     f.`Task Instance ID`,
# MAGIC     TRIM(f.`value`) AS value,
# MAGIC     row_number() OVER (PARTITION BY f.`Task Instance ID` ORDER BY f.`Task Instance ID`) AS RN
# MAGIC   FROM 
# MAGIC     StringSplit_Subject f
# MAGIC   WHERE f.RN = 1
# MAGIC );
# MAGIC
# MAGIC -- Remove post "-"
# MAGIC CREATE OR REPLACE TEMPORARY VIEW StringSplit_Subject_III AS
# MAGIC (
# MAGIC   SELECT 
# MAGIC     `Task Instance ID`,
# MAGIC     CASE
# MAGIC       WHEN t1.value LIKE 'Release Re-Inspection%' THEN t1.value
# MAGIC       WHEN t1.value LIKE '%R8-Request%' THEN t1.value
# MAGIC       WHEN t1.value LIKE '%Kick-off%' THEN t1.value
# MAGIC       WHEN t1.value LIKE '%Pre-maintenance%' THEN t1.value
# MAGIC       WHEN t1.value LIKE '%single-story%' THEN t1.value
# MAGIC       WHEN t1.value LIKE '%two-story%' THEN t1.value
# MAGIC       WHEN t1.value LIKE '%-y-%' THEN t1.value
# MAGIC       WHEN t1.value LIKE '%-yr-%' THEN t1.value
# MAGIC       --WHEN t1.value LIKE '%[0-9]-[0-9]%' THEN t1.value
# MAGIC       ELSE s.value
# MAGIC     END AS Value,
# MAGIC     row_number() OVER(PARTITION BY `Task Instance ID`, t1.value ORDER BY t1.value) AS RN
# MAGIC   FROM 
# MAGIC     StringSplit_Subject_II t1
# MAGIC     LATERAL VIEW explode(split(t1.Value, '-')) s AS value
# MAGIC );
# MAGIC
# MAGIC --Final Preprocessing
# MAGIC CREATE OR REPLACE TEMPORARY VIEW StringSplit_Subject_IIII AS
# MAGIC (
# MAGIC SELECT
# MAGIC     CASE
# MAGIC         WHEN INSTR(Value, 'NCS00') > 0 THEN LEFT(Value, (INSTR(Value, 'NCS00') -2))
# MAGIC     ELSE Value
# MAGIC     END AS Description
# MAGIC FROM StringSplit_Subject_III
# MAGIC WHERE Value <> ''
# MAGIC GROUP BY
# MAGIC     CASE
# MAGIC         WHEN INSTR(Value, 'NCS00') > 0 THEN LEFT(Value, (INSTR(Value, 'NCS00') -2))
# MAGIC     ELSE Value
# MAGIC     END
# MAGIC EXCEPT 
# MAGIC SELECT
# MAGIC     Description
# MAGIC FROM
# MAGIC     databricks_poc.tmr.dim_task_subject
# MAGIC );
# MAGIC
# MAGIC INSERT INTO databricks_poc.tmr.dim_task_subject
# MAGIC (`Description`)
# MAGIC SELECT Description FROM StringSplit_Subject_IIII
# MAGIC

# COMMAND ----------

# MAGIC %md
# MAGIC ## Populating Junk Task Dimensions

# COMMAND ----------

# MAGIC %sql
# MAGIC -- Checking for new values in dim_case_junk
# MAGIC
# MAGIC DELETE FROM databricks_poc.tmr.dim_casejunk
# MAGIC WHERE Code > 0;
# MAGIC
# MAGIC CREATE OR REPLACE TEMPORARY VIEW JunkCleanse AS
# MAGIC (
# MAGIC SELECT DISTINCT
# MAGIC     tmr.`BPM Reference Number`,
# MAGIC     tmr.`Applicant Name`,		-- Applicant
# MAGIC     tmr.`Applicant Company`,	-- Company
# MAGIC     tmr.`Development Name`,		-- Development
# MAGIC     tmr.`Site House Number`,	-- SiteHouseNumber
# MAGIC     tmr.`Site Building Name`,
# MAGIC     tmr.`Site Street 1`,
# MAGIC     tmr.`Site Street 2`,
# MAGIC     tmr.`Site Town`,
# MAGIC     tmr.`Site Post Code`,
# MAGIC     COALESCE(ct.`Connection Type`, 'WMD') AS `Connection Type`
# MAGIC FROM 
# MAGIC     databricks_poc.staging.tmr_raw tmr
# MAGIC     LEFT OUTER JOIN databricks_poc.staging.wmd_case_type ct
# MAGIC     ON tmr.`BPM Reference Number` = ct.`BPM Reference Number`
# MAGIC WHERE 
# MAGIC     tmr.`BPM Reference Number` <> ''
# MAGIC );
# MAGIC
# MAGIC CREATE OR REPLACE TEMPORARY VIEW JunkCleanse_II AS
# MAGIC (
# MAGIC SELECT 
# MAGIC     `BPM Reference Number`,
# MAGIC     `Applicant Name`,		-- Applicant
# MAGIC     `Applicant Company`,	-- Company
# MAGIC     `Development Name`,		-- Development
# MAGIC     `Site House Number`,	-- SiteHouseNumber
# MAGIC     `Site Building Name`,
# MAGIC     `Site Street 1`,
# MAGIC     `Site Street 2`,
# MAGIC     `Site Town`,
# MAGIC     `Site Post Code`,
# MAGIC     `Connection Type`
# MAGIC  FROM JunkCleanse
# MAGIC WHERE `Site Post Code` NOT LIKE '%Test%'
# MAGIC     OR `Applicant Name` NOT LIKE '% Test %'
# MAGIC     OR `Applicant Company` NOT LIKE '% Test %'
# MAGIC     OR `Site Town` NOT LIKE '% Test %'
# MAGIC     OR `BPM Reference Number` NOT IN ('PLS0027047')
# MAGIC );
# MAGIC
# MAGIC INSERT INTO databricks_poc.tmr.dim_casejunk
# MAGIC (
# MAGIC   `BPMReference`
# MAGIC   ,`Applicant`
# MAGIC   ,`Company`
# MAGIC   ,`Development`
# MAGIC   ,`SiteHouseNumber`
# MAGIC   ,`SiteBuildingName`
# MAGIC   ,`SiteStreet1`
# MAGIC   ,`SiteStreet2`
# MAGIC   ,`SiteTown`
# MAGIC   ,`SitePostcode`
# MAGIC   ,`WMDCaseType`
# MAGIC )  
# MAGIC SELECT 
# MAGIC     `BPM Reference Number`,
# MAGIC     `Applicant Name`,		-- Applicant
# MAGIC     `Applicant Company`,	-- Company
# MAGIC     `Development Name`,		-- Development
# MAGIC     `Site House Number`,	-- SiteHouseNumber
# MAGIC     `Site Building Name`,
# MAGIC     `Site Street 1`,
# MAGIC     `Site Street 2`,
# MAGIC     `Site Town`,
# MAGIC     UPPER(`Site Post Code`),
# MAGIC     `Connection Type`
# MAGIC FROM JunkCleanse_II
